package vip.liux.backend.application.serviceImpl.pay;

import com.google.common.annotations.VisibleForTesting;
import jakarta.annotation.Resource;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Lazy;
import org.springframework.http.MediaType;
import org.springframework.http.client.JdkClientHttpRequestFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.Assert;
import org.springframework.web.client.RestClient;
import vip.liux.application.dots.ListResultDto;
import vip.liux.application.dots.PagedResultDto;
import vip.liux.backend.application.service.pay.*;
import vip.liux.backend.application.service.pay.dto.app.PayAppDto;
import vip.liux.backend.application.service.pay.dto.notify.*;
import vip.liux.backend.application.service.pay.dto.order.PayOrderDto;
import vip.liux.backend.application.service.pay.dto.refund.PayRefundDetailsDto;
import vip.liux.backend.application.service.pay.dto.transfer.PayTransferDto;
import vip.liux.contracts.models.pay.PayNotifyLog;
import vip.liux.contracts.models.pay.PayNotifyTask;
import vip.liux.contracts.repositories.pay.PayNotifyLogRepository;
import vip.liux.contracts.repositories.pay.PayNotifyTaskRepository;
import vip.liux.contracts.shared.enums.pay.PayNotifyStatus;
import vip.liux.contracts.shared.enums.pay.PayNotifyType;
import vip.liux.core.utils.JsonUtils;
import vip.liux.lock.LockTemplate;

import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import static vip.liux.application.PageKit.toPage;
import static vip.liux.application.PageKit.toPaged;
import static vip.liux.backend.infrastructure.utils.ListJoinKit.backFill;

/**
 * 支付通知 Core Service 实现类
 *
 * @author 芋道源码
 */
@Service
@Slf4j
public class PayNotifyServiceImpl implements PayNotifyService, ApplicationContextAware {

    /**
     * 通知超时时间，单位：秒
     */
    public static final int NOTIFY_TIMEOUT = 120;
    /**
     * {@link #NOTIFY_TIMEOUT} 的毫秒
     */
    public static final long NOTIFY_TIMEOUT_MILLIS = 120 * 1000L;
    final RestClient restClient;
    @Resource
    @Lazy // 循环依赖，避免报错
    private PayOrderService orderService;
    @Resource
    @Lazy // 循环依赖，避免报错
    private PayAppService appService;
    @Resource
    @Lazy // 循环依赖，避免报错
    private PayTransferService transferService;
    @Resource
    @Lazy // 循环依赖，避免报错
    private PayRefundService refundService;
    @Resource
    private PayNotifyTaskRepository notifyTaskRepo;
    @Resource
    private PayNotifyLogRepository notifyLogRepo;
    @Resource
    private ThreadPoolTaskExecutor applicationTaskExecutor;
    @Resource
    private LockTemplate lockTemplate;
    private ApplicationContext applicationContext;


    public PayNotifyServiceImpl() {
        // 创建请求工厂，设置连接和读取超时时间（单位：毫秒）
        JdkClientHttpRequestFactory factory = new JdkClientHttpRequestFactory();
        factory.setReadTimeout((int) NOTIFY_TIMEOUT_MILLIS);

        // 构建 RestClient
        restClient = RestClient.builder().requestFactory(factory).build();
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void createPayNotifyTask(PayNotifyType type, String dataNo) {
        PayNotifyTask task = new PayNotifyTask().setType(type).setDataNo(dataNo).setLastExecuteTime(LocalDateTime.now()).setNextNotifyTime(LocalDateTime.now());
        task.setStatus(PayNotifyStatus.WAITING).setNextNotifyTime(LocalDateTime.now())
                .setNotifyTimes(0).setMaxNotifyTimes(PayNotifyTask.NOTIFY_FREQUENCY.length + 1);
        // 补充 appId + notifyUrl + merchant* 字段
        if (Objects.equals(task.getType(), PayNotifyType.ORDER)) {
            PayOrderDto order = orderService.getByOrderNo(task.getDataNo()); // 不进行非空判断，有问题直接异常
            task.setAppId(order.getAppId()).setNotifyUrl(order.getNotifyUrl())
                    .setMerchantOrderId(order.getMerchantOrderId());
        } else if (Objects.equals(task.getType(), PayNotifyType.REFUND)) {
            PayRefundDetailsDto refund = refundService.getRefundByRefundNo(task.getDataNo());
            task.setAppId(refund.getAppId()).setNotifyUrl(refund.getNotifyUrl())
                    .setMerchantOrderId(refund.getMerchantOrderId()).setMerchantRefundId(refund.getMerchantRefundId());
        } else if (Objects.equals(task.getType(), PayNotifyType.TRANSFER)) {
            PayTransferDto transfer = transferService.getTransferByNo(task.getDataNo());
            task.setAppId(transfer.getAppId()).setNotifyUrl(transfer.getNotifyUrl())
                    .setMerchantTransferId(transfer.getMerchantTransferId());
        }

        // 执行插入
        notifyTaskRepo.save(task);

        // 必须在事务提交后，在发起任务，否则 PayNotifyTaskDO 还没入库，就提前回调接入的业务
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
            @Override
            public void afterCommit() {
                // 异步的原因：避免阻塞当前事务，无需等待结果
                getSelf().executeNotifyAsync(task);
            }
        });
    }

    @Override
    public int executeNotify() throws InterruptedException {
        // 获得需要通知的任务
        List<PayNotifyTask> tasks = notifyTaskRepo.selectListByNotify();
        if (CollectionUtils.isEmpty(tasks)) {
            return 0;
        }

        // 遍历，逐个通知
        CountDownLatch latch = new CountDownLatch(tasks.size());
        tasks.forEach(task -> applicationTaskExecutor.execute(() -> {
            try {
                executeNotify(task);
            } finally {
                latch.countDown();
            }
        }));
        // 等待完成
        awaitExecuteNotify(latch);
        // 返回执行完成的任务数（成功 + 失败)
        return tasks.size();
    }

    /**
     * 等待全部支付通知的完成
     * 每 1 秒会打印一次剩余任务数量
     *
     * @param latch Latch
     * @throws InterruptedException 如果被打断
     */
    private void awaitExecuteNotify(CountDownLatch latch) throws InterruptedException {
        long size = latch.getCount();
        for (int i = 0; i < NOTIFY_TIMEOUT; i++) {
            if (latch.await(1L, TimeUnit.SECONDS)) {
                return;
            }
            log.info("[awaitExecuteNotify][任务处理中， 总任务数({}) 剩余任务数({})]", size, latch.getCount());
        }
        log.error("[awaitExecuteNotify][任务未处理完，总任务数({}) 剩余任务数({})]", size, latch.getCount());
    }

    /**
     * 异步执行单个支付通知
     *
     * @param task 通知任务
     */
    @Async
    public void executeNotifyAsync(PayNotifyTask task) {
        executeNotify(task);
    }

    /**
     * 【加锁】执行单个支付通知
     *
     * @param task 通知任务
     */
    @SneakyThrows
    public void executeNotify(PayNotifyTask task) {
        Assert.notNull(task.getId(), "任务 ID 不能为空");
        // 分布式锁，避免并发问题
        lockTemplate.execute(task.getId().toString(), lockKey -> {
            // 校验，当前任务是否已经被通知过
            // 虽然已经通过分布式加锁，但是可能同时满足通知的条件，然后都去获得锁。此时，第一个执行完后，第二个还是能拿到锁，然后会再执行一次。
            // 因此，此处我们通过第 notifyTimes 通知次数是否匹配来判断
            PayNotifyTask dbTask = notifyTaskRepo.findById(task.getId()).orElseThrow();
            if (ObjectUtils.notEqual(task.getNotifyTimes(), dbTask.getNotifyTimes())) {
                log.warn("[executeNotifySync][task({}) 任务被忽略，原因是它的通知不是第 ({}) 次，可能是因为并发执行了]",
                        JsonUtils.toJson(task), dbTask.getNotifyTimes());
            } else {
                // 执行通知
                getSelf().executeNotify0(dbTask);
            }

            return null;
        });
    }

    @Transactional(rollbackFor = Exception.class)
    public void executeNotify0(PayNotifyTask task) {
        // 发起回调
        PayNotifyResp invokeResult = null;
        Throwable invokeException = null;
        try {
            invokeResult = executeNotifyInvoke(task);
        } catch (Throwable e) {
            invokeException = e;
        }

        // 处理结果
        PayNotifyStatus newStatus = processNotifyResult(task, invokeResult, invokeException);

        // 记录 PayNotifyLog 日志
        String response = invokeException != null ? ExceptionUtils.getRootCauseMessage(invokeException) :
                JsonUtils.toJson(invokeResult);
        notifyLogRepo.save(new PayNotifyLog().setTaskId(task.getId())
                .setNotifyTimes(task.getNotifyTimes() + 1).setStatus(newStatus).setResponse(response));
    }

    /**
     * 执行单个支付任务的 HTTP 调用
     *
     * @param task 通知任务
     * @return HTTP 响应
     */
    private PayNotifyResp executeNotifyInvoke(PayNotifyTask task) {
        // 拼接 body 参数
        Object request;
        if (Objects.equals(task.getType(), PayNotifyType.ORDER)) {
            request = PayOrderNotifyDto.builder().merchantOrderId(task.getMerchantOrderId())
                    .payOrderNo(task.getDataNo()).build();
        } else if (Objects.equals(task.getType(), PayNotifyType.REFUND)) {
            request = PayRefundNotifyDto.builder().merchantOrderId(task.getMerchantOrderId())
                    .merchantRefundId(task.getMerchantRefundId())
                    .payRefundNo(task.getDataNo()).build();
        } else if (Objects.equals(task.getType(), PayNotifyType.TRANSFER)) {
            request = PayTransferNotifyDto.builder().merchantTransferId(task.getMerchantTransferId())
                    .payTransferNo(task.getDataNo()).build();
        } else {
            throw new RuntimeException("未知的通知任务类型：" + JsonUtils.toJson(task));
        }

        // 发起请求
        return RestClient.create().post().uri(task.getNotifyUrl())
                .body(JsonUtils.toJson(request))
                .contentType(MediaType.APPLICATION_JSON)
                .accept(MediaType.APPLICATION_JSON)
                .retrieve()
                .body(PayNotifyResp.class);
    }

    /**
     * 处理并更新通知结果
     *
     * @param task            通知任务
     * @param invokeResult    通知结果
     * @param invokeException 通知异常
     * @return 最终任务的状态
     */
    @VisibleForTesting
    PayNotifyStatus processNotifyResult(PayNotifyTask task, PayNotifyResp invokeResult, Throwable invokeException) {
        // 设置通用的更新 PayNotifyTask 的字段
        PayNotifyTask updateTask = new PayNotifyTask()
                .setLastExecuteTime(LocalDateTime.now())
                .setNotifyTimes(task.getNotifyTimes() + 1);

        // 情况一：调用成功
        if (invokeResult != null && invokeResult.isSuccess()) {
            updateTask.setStatus(PayNotifyStatus.SUCCESS);
            notifyTaskRepo.updateLastExecuteTimeAndNotifyTimesAndStatusById(
                    updateTask.getLastExecuteTime(), updateTask.getNotifyTimes(), updateTask.getStatus(), task.getId());
            return updateTask.getStatus();
        }

        // 情况二：调用失败、调用异常
        // 2.1 超过最大回调次数
        if (updateTask.getNotifyTimes() >= PayNotifyTask.NOTIFY_FREQUENCY.length) {
            updateTask.setStatus(PayNotifyStatus.FAILURE);
            notifyTaskRepo.updateLastExecuteTimeAndNotifyTimesAndStatusById(
                    updateTask.getLastExecuteTime(), updateTask.getNotifyTimes(), updateTask.getStatus(), task.getId());
            return updateTask.getStatus();
        }
        // 2.2 未超过最大回调次数
        updateTask.setNextNotifyTime(LocalDateTime.now()
                .plusSeconds(PayNotifyTask.NOTIFY_FREQUENCY[updateTask.getNotifyTimes()]));
        updateTask.setStatus(invokeException != null ? PayNotifyStatus.REQUEST_FAILURE
                : PayNotifyStatus.REQUEST_SUCCESS);
        notifyTaskRepo.updateLastExecuteTimeAndNotifyTimesAndStatusById(
                updateTask.getLastExecuteTime(), updateTask.getNotifyTimes(), updateTask.getStatus(), task.getId());
        return updateTask.getStatus();
    }

    @Override
    public PayNotifyTaskDto getNotifyTask(Long id) {
        PayNotifyTaskDto taskDto = AutoMapperProfile.INSTANCE.convert(notifyTaskRepo.findById(id).orElseThrow());
        Optional.ofNullable(taskDto)
                .ifPresent(o -> Optional.ofNullable(appService.getApp(o.getAppId())).ifPresent(p -> o.setAppName(p.getName())));
        return taskDto;
    }

    @Override
    public PagedResultDto<PayNotifyTaskDto> getNotifyTaskPage(PayNotifyTaskQueryDto input) {
        PagedResultDto<PayNotifyTaskDto> pagedResult = toPaged(notifyTaskRepo.findPage(
                input.getAppId(), input.getType(), input.getDataNo(),
                input.getStatus(), input.getMerchantOrderId(),
                input.getMerchantRefundId(), input.getMerchantTransferId(),
                input.getStart(), input.getEnd(), toPage(input)
        ).map(AutoMapperProfile.INSTANCE::convert));
        if (pagedResult.getTotalCount() > 0) {
            // 补充应用信息
            ListResultDto<PayAppDto> appList = appService.getAppList(pagedResult.getItems().stream().map(PayNotifyTaskDto::getAppId).toList());

            backFill(
                    pagedResult.getItems(),
                    appList.getItems(),
                    PayNotifyTaskDto::getAppId,
                    PayAppDto::getId,
                    (order, app) -> order.setAppName(app.getName())
            );
        }
        return pagedResult;
    }

    @Override
    public PayNotifyTaskDetailDto getNotifyTaskDetail(Long id) {
        List<PayNotifyTaskDetailDto.Log> logs = notifyLogRepo.findByTaskId(id).stream()
                .map(AutoMapperProfile.INSTANCE::convert)
                .toList();
        PayNotifyTaskDetailDto detailDto = AutoMapperProfile.INSTANCE.convert(notifyTaskRepo.findById(id).orElseThrow(), logs);
        Optional.ofNullable(detailDto)
                .ifPresent(o -> Optional.ofNullable(appService.getApp(o.getAppId())).ifPresent(p -> o.setAppName(p.getName())));
        return detailDto;
    }

    /**
     * 获得自身的代理对象，解决 AOP 生效问题
     *
     * @return 自己
     */
    private PayNotifyServiceImpl getSelf() {
        return applicationContext.getBean(getClass());
    }


    @Override
    public void setApplicationContext(@NotNull ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
